package com.taobao.analyse.map;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.analyse.utils.LCDataParser;

/**
 * 增量生命周期编辑记录mapper
 * @author xuanwei
 * @version 2012-3-15
 * @Mail-To xuanwei@taobao.com
 */
public class IncrMapper extends MapReduceBase implements
		Mapper<BytesWritable, BytesWritable, LongWritable, Text> {
	private DataInputBuffer	input	= new DataInputBuffer();

	@Override
	public void map(BytesWritable key, BytesWritable value, OutputCollector<LongWritable, Text> output,
			Reporter reporter) throws IOException {
		try {
			byte[] payload = new byte[value.getLength()];
			System.arraycopy(value.getBytes(), 0, payload, 0, payload.length);

			input.reset(payload, payload.length);
			Map<String, Object> val = LCDataParser.read(input, reporter);

			// do your thing
		} catch (Exception e) {
			reporter.incrCounter("LCParser", "MapperExe", 1);
		}
	}
}
